package com.bootdo.clouddoES.controller;



import com.bootdo.clouddocommon.context.FilterContextHandler;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;




@RestController
@RequestMapping("/pay")
public class TestController {
    @Autowired
    private RabbitMqProducerDemo producerDemo;
    @Autowired
    private RabbitMqConsumerDemo consumerDemo;
    @Autowired
    private InterfaceProducerDemo interfaceProducerDemo;
    @Autowired
    private InterfaceConsumerDemo interfaceConsumerDemo;
    @Autowired // 注入rabbitmq 模板
    private AmqpTemplate rabbitTemplate;
    @Autowired
    private KafkaTemplate kafkaTemplate;

    @GetMapping("1111")
    public String LoadPay() throws IOException {
        System.out.println("获取的token"+FilterContextHandler.getToken());
        producerDemo.sendHelloMessage();
        producerDemo.sendUserMessage();
        producerDemo.sendtest();
        return "11111";
    }

    @GetMapping("2222")
    public String LoadPay222() throws IOException {
   System.out.println();
        consumerDemo.executetest();
        return "11111";
    }
    @PostMapping("/sentDirectMessage")
    public String sentDirectMessage(){


        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", "1111");

        return "ok";
    }
    @GetMapping("/sendTopicMessage1")
    public String sendTopicMessage1() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: M A N ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> manMap = new HashMap<>();
        manMap.put("messageId", messageId);
        manMap.put("messageData", messageData);
        manMap.put("createTime", createTime);
        rabbitTemplate.convertAndSend("topicExchange", "topic.man", manMap);
        return "ok";
    }

    @GetMapping("/sendTopicMessage2")
    public String sendTopicMessage2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: woman is all ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> womanMap = new HashMap<>();
        womanMap.put("messageId", messageId);
        womanMap.put("messageData", messageData);
        womanMap.put("createTime", createTime);
        rabbitTemplate.convertAndSend("topicExchange", "topic.woman", womanMap);
        return "ok";
    }


    @PostMapping("/sendFanoutMessage")
    public String sendFanoutMessage() {
        String msgId= UUID.randomUUID().toString();
        String msgInfo="test fanout message";
        SimpleDateFormat df=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String createTime=df.format(new Date());

        Map<String,Object> map=new HashMap<String,Object>();
        map.put("msgId",msgId);
        map.put("msgInfo",msgInfo);
        map.put("createTime",createTime);

        rabbitTemplate.convertAndSend("fanoutExchange", null, map);

        System.out.println(map.toString());
        return "ok";
    }
    @GetMapping("/sendMsg/{message}")
    public String sendMsg(@PathVariable String message){
        rabbitTemplate.convertAndSend("X", "XA", "消息来自TTL为10s队列QA："+message);
        rabbitTemplate.convertAndSend("X", "XB", "消息来自TTL为40s队列QB："+message);
        return "发送成功";
    }
    @GetMapping("/interfacesendMsg")
    public String interfacesendMsg(@RequestParam String message){
        interfaceProducerDemo.sendInterfaceMessage();
        String aa = interfaceConsumerDemo.executeHello(message);
        return "发送成功"+aa;
    }
    @Autowired
    private KafkaProducer producer;

    @RequestMapping("/send/{topic}")
    public void send(@PathVariable String topic){
        producer.send(topic);

    }

    @GetMapping("/kafka/callbackTwo/{message}")
    public void sendMessage3(@PathVariable("message") String callbackMessage) {
        kafkaTemplate.send("topic1", callbackMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("发送消息失败："+ex.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, Object> result) {
                System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                        + result.getRecordMetadata().partition() + "-" + result.getRecordMetadata().offset());
            }
        });
    }


    @GetMapping("/kafka/transaction")
    public void sendMessage7() {
        // 声明事务：后面报错消息不会发出去
        kafkaTemplate.executeInTransaction(operations -> {
            operations.send("topic1", "test executeInTransaction");
            throw new RuntimeException("fail");
        });

        // 不声明事务：后面报错但前面消息已经发送成功了
        kafkaTemplate.send("topic1", "test executeInTransaction");
        throw new RuntimeException("fail");

    }
}
